package com.star.jersey.esJob.parser;

import com.dangdang.ddframe.job.config.JobCoreConfiguration;
import com.dangdang.ddframe.job.config.JobTypeConfiguration;
import com.dangdang.ddframe.job.config.dataflow.DataflowJobConfiguration;
import com.dangdang.ddframe.job.config.script.ScriptJobConfiguration;
import com.dangdang.ddframe.job.config.simple.SimpleJobConfiguration;
import com.dangdang.ddframe.job.event.rdb.JobEventRdbConfiguration;
import com.dangdang.ddframe.job.executor.handler.JobProperties;
import com.dangdang.ddframe.job.lite.config.LiteJobConfiguration;
import com.dangdang.ddframe.job.lite.spring.api.SpringJobScheduler;
import com.dangdang.ddframe.job.reg.zookeeper.ZookeeperRegistryCenter;
import com.star.jersey.esJob.annotation.ElasticJobConfig;
import com.star.jersey.esJob.config.EsJobZookeeperProperties;
import com.star.jersey.esJob.enums.ElasticJobTypeEnum;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.config.BeanDefinition;
import org.springframework.beans.factory.support.BeanDefinitionBuilder;
import org.springframework.beans.factory.support.DefaultListableBeanFactory;
import org.springframework.beans.factory.support.ManagedList;
import org.springframework.boot.context.event.ApplicationReadyEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;
import org.springframework.util.StringUtils;

import java.util.List;
import java.util.Map;

/**
 * @author zhouhengzhe
 * @description: es-job解析器
 * <p>
 * ApplicationReadyEvent ：事件尽可能晚发布，以指示应用程序已准备好为请求提供服务。该事件的来源是SpringApplication本身，但要注意修改其内部状态，
 * 因为届时所有初始化步骤都将完成。顾名思义就是SpringApplication准备就绪之后的事件，我们可以把一些有可能出现类未初始化异常的操作放在这里面执行
 *
 * 使用record特性可以减少get，构造方法
 * @date 2022/10/17 1:26
 * @since v1
 */
@Slf4j
public record ElasticJobConfParser(EsJobZookeeperProperties jobZookeeperProperties,
                                   ZookeeperRegistryCenter zookeeperRegistryCenter) implements ApplicationListener<ApplicationReadyEvent> {

    @Override
    public void onApplicationEvent(ApplicationReadyEvent event) {
        try {
            ApplicationContext applicationContext = event.getApplicationContext();
            Map<String, Object> beanMap = applicationContext.getBeansWithAnnotation(ElasticJobConfig.class);
            for (Object confBean : beanMap.values()) {
                Class<?> clazz = confBean.getClass();
                if (clazz.getName().indexOf("$") > 0) {
                    String className = clazz.getName();
                    clazz = Class.forName(className.substring(0, className.indexOf("$")));
                }
                // 	获取接口类型 用于判断是什么类型的任务
                String jobTypeName = clazz.getInterfaces()[0].getSimpleName();
                //	获取配置项 ElasticJobConfig
                ElasticJobConfig conf = clazz.getAnnotation(ElasticJobConfig.class);

                String eventTraceRdbDataSource = conf.eventTraceRdbDataSource();

                String jobName = getJobName(conf);

                //	先把当当网的esjob的相关configuration
                JobCoreConfiguration coreConfig = JobCoreConfiguration
                        .newBuilder(jobName, conf.cron(), conf.shardingTotalCount())
                        .shardingItemParameters(conf.shardingItemParameters())
                        .description(conf.description())
                        .failover(conf.failover())
                        .jobParameter(conf.jobParameter())
                        .misfire(conf.misfire())
                        .jobProperties(JobProperties.JobPropertiesEnum.JOB_EXCEPTION_HANDLER.getKey(), conf.jobExceptionHandler())
                        .jobProperties(JobProperties.JobPropertiesEnum.EXECUTOR_SERVICE_HANDLER.getKey(), conf.executorServiceHandler())
                        .build();
                //获取定时任务类型配置

                // LiteJobConfiguration
                LiteJobConfiguration jobConfig = LiteJobConfiguration
                        .newBuilder(getJobTypeConfiguration(jobTypeName, conf, clazz, coreConfig))
                        .overwrite(conf.overwrite())
                        .disabled(conf.disabled())
                        .monitorPort(conf.monitorPort())
                        .monitorExecution(conf.monitorExecution())
                        .maxTimeDiffSeconds(conf.maxTimeDiffSeconds())
                        .jobShardingStrategyClass(conf.jobShardingStrategyClass())
                        .reconcileIntervalMinutes(conf.reconcileIntervalMinutes())
                        .build();

                // 	创建一个Spring的beanDefinition
                BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(SpringJobScheduler.class);
                factory.setInitMethodName("init");
                factory.setScope("prototype");

                //	1.添加bean构造参数，相当于添加自己的真实的任务实现类
                if (!ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
                    factory.addConstructorArgValue(confBean);
                }
                //	2.添加注册中心
                factory.addConstructorArgValue(this.zookeeperRegistryCenter);
                //	3.添加LiteJobConfiguration
                factory.addConstructorArgValue(jobConfig);

                //	4.如果有eventTraceRdbDataSource 则也进行添加
                if (StringUtils.hasText(eventTraceRdbDataSource)) {
                    BeanDefinitionBuilder rdbFactory = BeanDefinitionBuilder.rootBeanDefinition(JobEventRdbConfiguration.class);
                    rdbFactory.addConstructorArgReference(eventTraceRdbDataSource);
                    factory.addConstructorArgValue(rdbFactory.getBeanDefinition());
                }

                //  5.添加监听
                List<?> elasticJobListeners = getTargetElasticJobListeners(conf);
                factory.addConstructorArgValue(elasticJobListeners);

                // 	接下来就是把factory 也就是 SpringJobScheduler注入到Spring容器中
                DefaultListableBeanFactory defaultListableBeanFactory = (DefaultListableBeanFactory) applicationContext.getAutowireCapableBeanFactory();

                String registerBeanName = conf.name() + "SpringJobScheduler";
                defaultListableBeanFactory.registerBeanDefinition(registerBeanName, factory.getBeanDefinition());
                SpringJobScheduler scheduler = (SpringJobScheduler) applicationContext.getBean(registerBeanName);
                scheduler.init();
                log.info("启动elastic-job作业: " + jobName);
            }
            log.info("共计启动elastic-job作业数量为: {} 个", beanMap.values().size());

        } catch (Exception e) {
            log.error("elastic-job 启动异常, 系统强制退出", e);
            System.exit(1);
        }
    }

    /**
     * 获取定时任务名字
     *
     * @param conf es-job配置
     * @return
     */
    private String getJobName(ElasticJobConfig conf) {
        return this.jobZookeeperProperties.getNamespace() + "." + conf.name();
    }

    /**
     * 我到底要创建什么样的任务.
     *
     * @param jobTypeName 任务类型
     * @param conf        es-job配置
     * @param clazz       类对象
     * @param coreConfig  作业核心配置
     * @return
     */
    private JobTypeConfiguration getJobTypeConfiguration(String jobTypeName, ElasticJobConfig conf, Class<?> clazz, JobCoreConfiguration coreConfig) {
        //	我到底要创建什么样的任务.
        JobTypeConfiguration typeConfig = null;

        if (ElasticJobTypeEnum.SIMPLE.getType().equals(jobTypeName)) {
            typeConfig = new SimpleJobConfiguration(coreConfig, clazz.getName());
        }

        if (ElasticJobTypeEnum.DATAFLOW.getType().equals(jobTypeName)) {
            typeConfig = new DataflowJobConfiguration(coreConfig, clazz.getName(), conf.streamingProcess());
        }

        if (ElasticJobTypeEnum.SCRIPT.getType().equals(jobTypeName)) {
            typeConfig = new ScriptJobConfiguration(coreConfig, conf.scriptCommandLine());
        }
        return typeConfig;
    }

    private List<BeanDefinition> getTargetElasticJobListeners(ElasticJobConfig conf) {
        List<BeanDefinition> result = new ManagedList<BeanDefinition>(2);
        String listeners = conf.listener();
        if (StringUtils.hasText(listeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(listeners);
            factory.setScope("prototype");
            result.add(factory.getBeanDefinition());
        }

        String distributedListeners = conf.distributedListener();
        long startedTimeoutMilliseconds = conf.startedTimeoutMilliseconds();
        long completedTimeoutMilliseconds = conf.completedTimeoutMilliseconds();

        if (StringUtils.hasText(distributedListeners)) {
            BeanDefinitionBuilder factory = BeanDefinitionBuilder.rootBeanDefinition(distributedListeners);
            factory.setScope("prototype");
            factory.addConstructorArgValue(startedTimeoutMilliseconds);
            factory.addConstructorArgValue(completedTimeoutMilliseconds);
            result.add(factory.getBeanDefinition());
        }
        return result;
    }

}
